package internal

import (
	"fmt"
	"hash/crc32"
	"t11Lib/t11Errors"
	"t11Lib/t11Log"
	"t11Lib/t11Util"
	"time"
	"亚太-iot-服务器/biz"
	"亚太-iot-服务器/conf"
)

import (
	"context"
	mqtt2 "github.com/eclipse/paho.mqtt.golang"
)

const MqttThreadCnt = 5
const MqttChanSize = 10000

//实现对mqtt的监听与发送，将mqtt数据重定向到chan上
type MqttListenerManagerInternal struct {
	sendChans  chan *biz.MqMsg
	recvChans  chan *biz.MqMsg
	msgHandler func(msg *biz.MqMsg) t11Errors.T11Error
}

func (self *MqttListenerManagerInternal) RegisterMsgHandler(handler func(msg *biz.MqMsg) t11Errors.T11Error) {
	self.msgHandler = handler
}

func ClientIdToInt(clientId string) uint32 {
	return crc32.ChecksumIEEE([]byte(clientId)) % MqttThreadCnt
}

func (self *MqttListenerManagerInternal) SendData(clientId string, data string) t11Errors.T11Error {
	mqMsg := &biz.MqMsg{
		ClientId: clientId,
		Topic:    fmt.Sprintf("/%v/toDev/%v", ClientIdToInt(clientId), clientId),
		Data:     data,
	}
	select {
	case self.sendChans <- mqMsg:
		return nil
	default:
		return t11Errors.New("send failed")
	}
}

func CreateMqttThread() *MqttListenerManagerInternal {
	mgr := &MqttListenerManagerInternal{}
	threadCnt := MqttThreadCnt
	mgr.recvChans = make(chan *biz.MqMsg, MqttChanSize)
	mgr.sendChans = make(chan *biz.MqMsg, MqttChanSize)
	for i := 0; i < threadCnt; i++ {
		go func(i int) {
			for {
				err := mgr.mqttSendLoop(i)
				t11Log.Debuglnf("mqttSendLoop 失败，err = %v", err)
				time.Sleep(time.Second * 5)
			}
		}(i)
		go func() {
			mgr.recvLoop()
		}()
	}
	return mgr
}

func (self *MqttListenerManagerInternal) recvLoop() {
	for {
		select {
		case msg := <-self.recvChans:
			func() {
				defer func() {
					if r := recover(); r != nil {

					}
				}()
				msgHandler := self.msgHandler
				if msgHandler != nil {
					t11Log.Debuglnf("处理mqtt数据包：%+v", msg)
					if err := msgHandler(msg); err != nil {
						t11Log.Debuglnf("err = %v", err)
					}
				}
			}()
		}
	}
}

func (self *MqttListenerManagerInternal) mqttSendLoop(clientIdx int) t11Errors.T11Error {
	defer recover()
	var f mqtt2.MessageHandler = func(client mqtt2.Client, msg mqtt2.Message) {
		data := &biz.MqMsg{
			Topic: msg.Topic(),
			Data:  string(msg.Payload()),
		}
		select {
		case self.recvChans <- data:
		default:
			return
		}
	}
	listenTopic := fmt.Sprintf("/%v/fromDev/#", clientIdx)
	mqttUrl := fmt.Sprintf("tcp://%v:%v", conf.AppConfig.MqttHost, conf.AppConfig.MqttPort)
	opts := mqtt2.NewClientOptions().AddBroker(mqttUrl)
	opts.SetClientID(fmt.Sprintf("%v-admin-%v",t11Util.RandMd5(), clientIdx))
	opts.SetUsername(conf.AppConfig.MqttAdminUser)
	opts.SetPassword(conf.AppConfig.MqttAdminPassword)
	opts.SetDefaultPublishHandler(f)
	ctx, cancel := context.WithCancel(context.TODO())
	opts.SetConnectionLostHandler(func(cc mqtt2.Client, e error) {
		clientId := ""
		if cc != nil {
			rd := cc.OptionsReader()
			clientId = rd.Username()
		}
		t11Log.Debuglnf("mqtt连接关闭，client = %v, error = %v", clientId, e)
		cancel()
	})
	c := mqtt2.NewClient(opts)
	if token := c.Connect(); token.Wait() && token.Error() != nil {
		t11Log.Debuglnf("mqtt连接失败, token = %v", token)
		return t11Errors.Wrap(token.Error())
	}
	defer c.Disconnect(250)
	t11Log.Debuglnf("mqtt链接成功, listenTopic = %v", listenTopic)

	if token := c.Subscribe(listenTopic, 0, nil); token.Wait() && token.Error() != nil {
		return t11Errors.Wrap(token.Error())
	}
	defer func() {
		if token := c.Unsubscribe(listenTopic); token.Wait() && token.Error() != nil {
			t11Log.Debuglnf("mqtt失败：%v", t11Errors.Wrap(token.Error()))
		}
	}()

	for {
		select {
		case <-ctx.Done():
			return t11Errors.New("canceled")
		case msg := <-self.sendChans:
			token := c.Publish(msg.Topic, 0, false, msg.Data)
			if !token.WaitTimeout(time.Second * 10) {
				return t11Errors.New("发送失败")
			}
		}
	}
}
